package cn.anytrust.hbase.service.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.anytrust.hbase.ORMHBaseTable;
import cn.anytrust.hbase.service.HBaseConnectionService;
import cn.anytrust.hbase.util.HBaseUtil;

/**
 * 
 * @author gaoweigong
 * @createtime 2016-01-29
 */
public class HBaseConnectionServiceImpl implements HBaseConnectionService {
	public Logger logger = LoggerFactory
			.getLogger(HBaseConnectionServiceImpl.class);
	private Configuration config;

	/**
	 * 
	 */
	public HBaseConnectionServiceImpl() {
	}

	/**
	 * 获取配置，当前使用默认配置
	 * 
	 * @return the config
	 */
	public Configuration getConfig() {
		if (config == null) {
			config = HBaseConfiguration.create();
		}
		return config;
	}

	@Override
	public List<Object> get(Object obj, String... rowkeys) {
		String tableName = getORMTable(obj);
		if (StringUtils.isBlank(tableName)) {
			return null;
		}
		List<Result> results = getResults(tableName, rowkeys);
		List<Object> objs = new ArrayList<Object>();
		if (results.isEmpty()) {
			return null;
		}
		for (Result result : results) {
			if (result == null || result.isEmpty()) {
				continue;
			}
			try {
				obj = HBaseUtil.result2Bean(result, obj);
			} catch (Exception e) {
				logger.warn("",e);
			}
			objs.add(obj);
		}
		return objs;
	}

	@Override
	public void delete(String tableName, String... rowkeys) {
		List<Delete> deletes = new ArrayList<Delete>();
		for (String rowkey : rowkeys) {
			if (StringUtils.isBlank(rowkey)) {
				continue;
			}
			deletes.add(new Delete(Bytes.toBytes(rowkey)));
		}
		delete(deletes, tableName);
	}

	@Override
	public void delete(Object obj, String... rowkeys) {
		String tableName = "";
		tableName = getORMTable(obj);
		if (StringUtils.isBlank(tableName)) {
			return;
		}
		List<Delete> deletes = new ArrayList<Delete>();
		for (String rowkey : rowkeys) {
			if (StringUtils.isBlank(rowkey)) {
				continue;
			}
			deletes.add(new Delete(Bytes.toBytes(rowkey)));
		}
		delete(deletes, tableName);

	}

	@Override
	public void insert(Object... objs) {
		List<Put> puts = new ArrayList<Put>();
		String tableName = "";
		for (Object obj : objs) {
			if (obj == null) {
				continue;
			}
			tableName = getORMTable(obj);
			try {
				Put put = HBaseUtil.bean2Put(obj);
				puts.add(put);
			} catch (Exception e) {
			}
		}
		savePut(puts, tableName);
	}

	private void delete(List<Delete> deletes, String TableName) {
		HConnection connection = null;
		HTableInterface table = null;
		try {
			if (StringUtils.isBlank(TableName)) {
				return;
			}
			connection = HConnectionManager.createConnection(getConfig());
			table = connection.getTable(TableName);
			table.delete(deletes);
		} catch (IOException e) {
		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (IOException e) {
				}
			}
			if (connection != null) {
				try {
					connection.close();
				} catch (IOException e) {
				}
			}
		}
	}

	private void savePut(List<Put> puts, String TableName) {
		HConnection connection = null;
		HTableInterface table = null;
		try {
			if (StringUtils.isBlank(TableName)) {
				return;
			}
			connection = HConnectionManager.createConnection(getConfig());
			table = connection.getTable(TableName);
			table.put(puts);
		} catch (IOException e) {
		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (IOException e) {
				}
			}
			if (connection != null) {
				try {
					connection.close();
				} catch (IOException e) {
				}
			}
		}
	}

	public List<Result> getResults(String tableName, String... rowkeys) {
		List<Result> resultList = new ArrayList<Result>();
		List<Get> gets = new ArrayList<Get>();
		for (String rowkey : rowkeys) {
			if (StringUtils.isBlank(rowkey)) {
				continue;
			}
			Get get = new Get(Bytes.toBytes(rowkey));
			gets.add(get);
		}
		HConnection connection = null;
		HTableInterface table = null;
		try {
			connection = HConnectionManager.createConnection(getConfig());
			table = connection.getTable(tableName);
			Result[] results = table.get(gets);
			Collections.addAll(resultList, results);
			return resultList;
		} catch (IOException e) {
			logger.warn("",e);
			return null;
		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (IOException e) {
				}
			}
			if (connection != null) {
				try {
					connection.close();
				} catch (IOException e) {
				}
			}
		}

	}

	private String getORMTable(Object obj) {
		ORMHBaseTable table = obj.getClass().getAnnotation(ORMHBaseTable.class);
		return table.tableName();
	}
}
